package com.atguigu.gmall.app.dwd.db;

import com.atguigu.gmall.utils.KafkaUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeRefundPaySuc {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop100:8020/edu");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // 设置TTL时间
        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.exec.state.ttl", "5 s");

        // TODO 3. 读取 Kafka dwd_trade_order_detail 主题数据，封装为 Flink SQL 表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "province_id string,\n" +
                "create_time string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "session_id string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "ts string\n" +
                ")" + KafkaUtil.getKafkaDDL("dwd_trade_order_detail", "dwd_trade_refund_pay_suc"));

        // TODO 4. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`ts` string\n" +
                ")" + KafkaUtil.getKafkaDDL("topic_db", "dwd_trade_refund_pay_suc"));

        // TODO 5. 筛选退款成功数据
        Table refundPayment = tableEnv.sqlQuery("select\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['payment_type'] payment_type,\n" +
                "data['callback_time'] callback_time,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `table` = 'payment_info'\n" +
                "and `type` = 'update'\n" +
                "and data['payment_status']='1006'");
        tableEnv.createTemporaryView("refund_payment", refundPayment);


        // TODO 6. 关联 2 张表获得退款成功宽表
        Table resultTable = tableEnv.sqlQuery("" +
                "select\n" +
                "od.id order_detail_id,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.province_id,\n" +
                "od.create_time,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.session_id,\n" +
                "pi.payment_type payment_type_code,\n" +
                "pi.callback_time,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +
                "pi.ts\n" +
                "from dwd_trade_order_detail od\n" +
                "join payment_info pi\n" +
                "on pi.order_id = od.order_id");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 7. 创建 dwd_trade_refund_pay 表
        tableEnv.executeSql("create table dwd_trade_refund_pay_suc(\n" +
                "order_detail_id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "province_id string,\n" +
                "create_time string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "session_id string,\n" +
                "payment_type_code string,\n" +
                "callback_time string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "ts string,\n" +
                "primary key(order_detail_id) not enforced\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_refund_pay_suc"));

        // TODO 8. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_refund_pay_suc select * from result_table");
    }
}
